package com.shujia.airPM25;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

class MyMapper extends Mapper<LongWritable, Text, Text, LongWritable> {
    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, LongWritable>.Context context) throws IOException, InterruptedException {
        //将一行的数据转成String类型
        String line = value.toString();
        String[] strings = line.split(",");
        //过滤出PM2.5对应的数据
        if (strings.length >= 4 && "PM2.5".equals(strings[2])) {
            for (int i = 3, j = 1001; i < strings.length; i++, j++) {
                //对一行数据做简单的清洗，因为有的时间没有监控到PM2.5的值
                if ("".equals(strings[i]) || strings[i] == null || " ".equals(strings[i])) {
                    strings[i] = "0";
                }

                context.write(new Text("date: " + strings[0] + "-城市编号:" + j), new LongWritable(Long.parseLong(strings[i])));
            }
        }
    }
}
